package com.cl.spark.config;

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.sql.SparkSession;
import org.springframework.context.annotation.Bean;
import org.springframework.stereotype.Component;

import java.io.IOException;

@Component
public class SparkConfig {
    @Bean(name = "sparkConf")
    public SparkConf sparkConf() {
        SparkConf conf = new SparkConf()
                .setAppName("SpringBootSparkApp")
                .setMaster("local[*]")
                .set("spark.executor.memory", "4g") // 每个Executor使用的内存，根据集群节点内存大小进行调整
                .set("spark.executor.cores", "4") // 每个Executor使用的CPU核心数，根据集群节点CPU核心数进行调整
                .set("spark.executor.instances", "10") // Executor的数量，根据集群规模和任务需求进行调整
                .set("spark.default.parallelism", "1000") // 设置默认的并行度，可以根据数据集大小和集群规模进行调整
                .set("spark.shuffle.partition", "100") // Shuffle操作的分区数，可以根据集群规模和任务特性进行调整
                .set("spark.memory.fraction", "0.6") // 用于执行和存储的内存占Spark执行器总内存的比例
                .set("spark.memory.storageFraction", "0.5") // 在执行器内存中，用于存储持久化RDDs的内存比例
                .set("spark.serializer", "org.apache.spark.serializer.KryoSerializer") // 使用Kryo序列化器来提高序列化性能
                .set("spark.rdd.compress", "true") // 启用RDD压缩，减少网络传输和存储开销
                .set("spark.sql.shuffle.partitions", "100") // SQL查询中shuffle操作的分区数
//                .set("spark.network.timeout", "300s") // 设置网络操作的超时时间
                .set("spark.speculation", "true"); // 开启推测执行，提高任务执行效率
                // 垃圾回收配置（可选，根据具体情况调整）
//                .set("spark.executor.extraJavaOptions", "-XX:+UseG1GC -XX:InitiatingHeapOccupancyPercent=35 -XX:MaxGCPauseMillis=100");
//                .set("fs.defaultFS", "hdfs://localhost:9000");  // 在本地运行，使用所有可用核心

        // 其他配置项...
        return conf;
    }

    @Bean(name = "sparkContext")
    public JavaSparkContext sparkContext(SparkSession sparkSession) {
        return JavaSparkContext.fromSparkContext(sparkSession.sparkContext());
    }

    @Bean(name = "SparkSession")
    public SparkSession SparkSession(SparkConf conf) {
        return SparkSession.builder().config(conf).getOrCreate();
    }

//    @Bean(name = "fileSystem")
//    public FileSystem fileSystem() {
//        Configuration conf = new Configuration();
//        // 获取Hadoop文件系统实例
//        try {
//            FileSystem fs = FileSystem.get(conf);
//            return fs;
//        } catch (IOException e) {
//            throw new RuntimeException(e);
//        }
//    }
}